これはインタラクティブなノートブックです。ローカルで実行するか、以下のリンクを使用できます：

​
Weaveとの統合：本番ダッシュボード

GenAIツールの状況は急速に進化しています - 新しいフレームワーク、ツール、アプリケーションが次々と登場しています。WeaveはあなたのすべてのGenAIモニタリングと評価のニーズに対応するワンストップショップを目指しています。これはまた、既存のプラットフォームと統合したり、プロジェクトや組織の特定のニーズに合わせてWeaveを拡張したりする必要がある場合もあることを意味します。 このクックブックでは、WeaveのパワフルなAPIと機能を活用して、Weaveのトレースビューの拡張として本番環境モニタリング用のカスタムダッシュボードを作成する方法を紹介します。以下に焦点を当てます：
  • Weaveからトレース、コスト、フィードバック、その他のメトリクスを取得する
  • ユーザーフィードバックとコスト分布の集計ビューを作成する
  • トークン使用量とレイテンシーの時間経過に伴う可視化を作成する
streamlitをインストールして実行することで、あなた自身のWeaveプロジェクトでこのダッシュボードを試すことができますこの本番環境ダッシュボードスクリプト! Example Production Dashboard with Weave

​
1. セットアップ

このチュートリアルを進めるには、以下のパッケージをインストールするだけで済みます： 
!pip install streamlit pandas plotly weave

​
2. 実装

​
2.1 Weaveクライアントの初期化とコストの定義

まず、Weaveクライアントを初期化し、各モデルのコストを追加する関数を設定します。
  • 多くの標準モデルの標準コストを含めていますが、独自のカスタムコストやカスタムモデルを簡単に追加することもできます。以下では、いくつかのモデルにカスタムコストを追加し、残りには標準コストを使用する方法を示します。
  • コストはWeaveで追跡される各呼び出しのトークンに基づいて計算されます。多くのLLMベンダーライブラリでは、トークン使用量を自動的に追跡しますが、任意の呼び出しにカスタムトークン数を返すことも可能です。カスタムモデルのトークン数とコスト計算を定義する方法については、このクックブックを参照してください - カスタムコストクックブック.
PROJECT_NAME = "wandb-smle/weave-cookboook-demo"
python
import weave

MODEL_NAMES = [
    # model name, prompt cost, completion cost
    ("gpt-4o-2024-05-13", 0.03, 0.06),
    ("gpt-4o-mini-2024-07-18", 0.03, 0.06),
    ("gemini/gemini-1.5-flash", 0.00025, 0.0005),
    ("gpt-4o-mini", 0.03, 0.06),
    ("gpt-4-turbo", 0.03, 0.06),
    ("claude-3-haiku-20240307", 0.01, 0.03),
    ("gpt-4o", 0.03, 0.06),
]

def init_weave_client(project_name):
    try:
        client = weave.init(project_name)
        for model, prompt_cost, completion_cost in MODEL_NAMES:
            client.add_cost(
                llm_id=model,
                prompt_token_cost=prompt_cost,
                completion_token_cost=completion_cost,
            )
    except Exception as e:
        print(f"Failed to initialize Weave client for project '{project_name}': {e}")
        return None
    else:
        return client

client = init_weave_client(PROJECT_NAME)

​
2.2 Weaveから呼び出しデータを取得する

Weaveから呼び出しデータを取得するには、2つの選択肢があります：
  1. 呼び出しごとにデータを取得する
  2. 高レベルAPIを使用する

​
2.2.1 呼び出しごとにデータを取得する

Weaveからデータにアクセスする最初の選択肢は、フィルタリングされた呼び出しのリストを取得し、呼び出しごとに必要なデータを抽出することです。そのためにcalls_query_stream APIを使用してWeaveから呼び出しデータを取得できます：
  • calls_query_stream API: This API allows us to fetch the calls data from Weave.
  • filter dictionary: This dictionary contains the filter parameters to fetch the calls data - see here で詳細を確認できます。
  • expand_columns list: This list contains the columns to expand in the calls data.
  • sort_by list: This list contains the sorting parameters for the calls data.
  • include_costs boolean: This boolean indicates whether to include the costs in the calls data.
  • include_feedback boolean: This boolean indicates whether to include the feedback in the calls data.
import itertools
from datetime import datetime, timedelta

import pandas as pd

def fetch_calls(client, project_id, start_time, trace_roots_only, limit):
    filter_params = {
        "project_id": project_id,
        "filter": {"started_at": start_time, "trace_roots_only": trace_roots_only},
        "expand_columns": ["inputs.example", "inputs.model"],
        "sort_by": [{"field": "started_at", "direction": "desc"}],
        "include_costs": True,
        "include_feedback": True,
    }
    try:
        calls_stream = client.server.calls_query_stream(filter_params)
        calls = list(
            itertools.islice(calls_stream, limit)
        )  # limit the number of calls to fetch if too many
        print(f"Fetched {len(calls)} calls.")
    except Exception as e:
        print(f"Error fetching calls: {e}")
        return []
    else:
        return calls

calls = fetch_calls(client, PROJECT_NAME, datetime.now() - timedelta(days=1), True, 100)
python
# the raw data is a list of Call objects
pd.DataFrame([call.dict() for call in calls]).head(3)
Weaveからの戻り値を使って呼び出しを処理するのは非常に簡単です - 関連情報を抽出し、辞書のリストに保存します。その後、辞書のリストをpandas DataFrameに変換して返します。 
import json
from datetime import datetime

import pandas as pd

def process_calls(calls):
    records = []
    for call in calls:
        feedback = call.summary.get("weave", {}).get("feedback", [])
        thumbs_up = sum(
            1
            for item in feedback
            if isinstance(item, dict) and item.get("payload", {}).get("emoji") == "👍"
        )
        thumbs_down = sum(
            1
            for item in feedback
            if isinstance(item, dict) and item.get("payload", {}).get("emoji") == "👎"
        )
        latency = call.summary.get("weave", {}).get("latency_ms", 0)

        records.append(
            {
                "Call ID": call.id,
                "Trace ID": call.trace_id,  # this is a unique ID for the trace that can be used to retrieve it
                "Display Name": call.display_name,  # this is an optional name you can set in the UI or programatically
                "Latency (ms)": latency,
                "Thumbs Up": thumbs_up,
                "Thumbs Down": thumbs_down,
                "Started At": pd.to_datetime(getattr(call, "started_at", datetime.min)),
                "Inputs": json.dumps(call.inputs, default=str),
                "Outputs": json.dumps(call.output, default=str),
            }
        )
    return pd.DataFrame(records)
python
df_calls = process_calls(calls)
df_calls.head(3)

​
2.2.2 高レベルAPIを使用する

すべての呼び出しを調べる代わりに、Weaveはモデルコスト、フィードバック、その他のメトリクスに直接アクセスするための高レベルAPIも提供しています。 例えば、コストについては、query_costs APIを使用してプロジェクトで使用されているすべてのLLMのコストを取得します： 
# Use cost API to get costs
costs = client.query_costs()
df_costs = pd.DataFrame([cost.dict() for cost in costs])
df_costs["total_cost"] = (
    df_costs["prompt_token_cost"] + df_costs["completion_token_cost"]
)

# only show the first row for every unqiue llm_id
df_costs

​
2.4 入力の収集と可視化の生成

次に、plotlyを使用して可視化を生成できます。これは最も基本的なダッシュボードですが、お好みにカスタマイズできます！より複雑な例については、Streamlitの例をhereで確認してください。 
import plotly.express as px
import plotly.graph_objects as go

def plot_feedback_pie_chart(thumbs_up, thumbs_down):
    fig = go.Figure(
        data=[
            go.Pie(
                labels=["Thumbs Up", "Thumbs Down"],
                values=[thumbs_up, thumbs_down],
                marker={"colors": ["#66b3ff", "#ff9999"]},
                hole=0.3,
            )
        ]
    )
    fig.update_traces(textinfo="percent+label", hoverinfo="label+percent")
    fig.update_layout(showlegend=False, title="Feedback Summary")
    return fig

def plot_model_cost_distribution(df):
    fig = px.bar(
        df,
        x="llm_id",
        y="total_cost",
        color="llm_id",
        title="Cost Distribution by Model",
    )
    fig.update_layout(xaxis_title="Model", yaxis_title="Cost (USD)")
    return fig

# See the source code for all the plots
python
plot_feedback_pie_chart(df_calls["Thumbs Up"].sum(), df_calls["Thumbs Down"].sum())
python
plot_model_cost_distribution(df_costs)

​
結論

このクックブックでは、WeaveのAPIと機能を使用してカスタム本番環境モニタリングダッシュボードを作成する方法を紹介しました。Weaveは現在、データの簡単な入力とカスタムプロセス用のデータ抽出のための迅速な統合に焦点を当てています。
  • データ入力：
    • フレームワークに依存しないトレースを@weave-op()デコレータで行い、CSVから呼び出しをインポートする可能性もあります（関連するimport cookbookを参照）
    • 様々なプログラミングフレームワークや言語からWeaveにログを記録するためのサービスAPIエンドポイント、詳細はhereを参照してください。
  • データ出力：
    • CSV、TSV、JSONL、JSON形式でのデータの簡単なダウンロード - 詳細はhereを参照してください。
    • プログラムによるデータへのアクセスを使用した簡単なエクスポート - このクックブックで説明されているエクスポートパネルの「Use Python」セクションを参照してください。詳細はhereを参照してください。
このカスタムダッシュボードはWeaveのネイティブトレースビューを拡張し、本番環境でのLLMアプリケーションのカスタマイズされたモニタリングを可能にします。より複雑なダッシュボードに興味がある場合は、あなた自身のWeaveプロジェクトURLを追加できるStreamlitの例をin this repoで確認してください。